//package com.hu.neo4j
//
//import java.sql.{Connection, DriverManager, PreparedStatement, ResultSetMetaData, SQLException, Time, Timestamp, Types}
//import java.util.Objects
//
//import com.hu.util.DateFormatUtils
//import org.apache.flink.api.java.utils.ParameterTool
//
////import java.sql.{Connection, PreparedStatement, ResultSetMetaData, Types}
//import java.text.SimpleDateFormat
//import java.util.StringJoiner
//
//import com.iyunbao.dc.common.metadata.JobConstant
//import com.iyunbao.dc.common.util.DateFormatUtils
//import com.iyunbao.dc.util.ODPSConnectUtil
//import org.apache.flink.api.common.io.RichOutputFormat
//import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
//import org.apache.flink.api.java.ExecutionEnvironment
//import org.apache.flink.api.java.typeutils.RowTypeInfo
//import org.apache.flink.configuration.Configuration
//import org.apache.flink.connector.jdbc.JdbcInputFormat
//import org.apache.flink.types.Row
//import org.apache.flink.util.Preconditions
//import org.slf4j.LoggerFactory
//
//object ODPSToNeo4jJob {
//
//  def main(args: Array[String]): Unit = {
//    val params: ParameterTool = ParameterTool.fromArgs(args)
//    val sqlQuery = params.get("sqlQuery")
//    val nodeName = params.get("nodeName")
//    val batchNum = params.getInt("batchNum")
//
//    val start = System.currentTimeMillis()
//    syncODPSData2Neo4j(sqlQuery, nodeName, batchNum)
//    val end = System.currentTimeMillis()
//    println("导入总耗时:" + (end - start))
//  }
//
//  def syncODPSData2Neo4j(sqlQuery: String, nodeName: String, batchNum: Integer): Unit = {
//    Preconditions.checkNotNull(sqlQuery, "query sql should not be null.");
//    val bizDate = DateFormatUtils.unixToDateTimeFormat(System.currentTimeMillis() - 86400000, "yyyyMMdd", false)
//    val simpleSql = formatSQl(sqlQuery, bizDate, true)
//    val sql = formatSQl(sqlQuery, bizDate, false)
//
//    //connection连接查询odps数据结果获取数据类型
//    val resultSet = ODPSConnectUtil.getODPSConn().prepareStatement(simpleSql).executeQuery()
//    val metaData = resultSet.getMetaData()
//    val colCount = metaData.getColumnCount
//
//    val odpsUrl = s"jdbc:odps:${JobConstant.DEFAULT_ODPS_END_POINT}?project=${JobConstant.DEFAULT_ODPS_PROJECT_NAME}"
//    val neo4jUrl = s"jdbc:neo4j:bolt://${JobConstant.DEFAULT_NEO4J_SERVER}:${JobConstant.DEFAULT_NEO4J_PORT}/"
//
//    //构建odps字段类型
//    val odpsFieldTypes = new Array[TypeInformation[_]](colCount)
//    for (i <- 1 to colCount) {
//      val mappingTuple = jdbcMappingBasicType(metaData.getColumnTypeName(i))
//      odpsFieldTypes(i - 1) = mappingTuple._1
//    }
//
//    val rowTypeInfo = new RowTypeInfo(odpsFieldTypes: _*)
//    val odpsInputFormat = JdbcInputFormat.buildJdbcInputFormat()
//      .setDrivername(JobConstant.ODPS_JDBC_DRIVER)
//      .setDBUrl(odpsUrl)
//      .setUsername(JobConstant.DEFAULT_ACCESSKEY_ID)
//      .setPassword(JobConstant.DEFAULT_ACCESSKEY_SECRET)
//      .setQuery(sql)
//      .setRowTypeInfo(rowTypeInfo)
//      .finish()
//
//    val env = ExecutionEnvironment.getExecutionEnvironment
//    //    env.setParallelism(1)
//
//    val value = env.createInput(odpsInputFormat)
//    val neo4jCql = odpsDataToNeo4jCQL(nodeName, metaData)
//    value.output(new Neo4jOutputFormat(neo4jUrl, JobConstant.DEFAULT_NEO4J_USER, JobConstant.DEFAULT_NEO4J_PASSWORD, neo4jCql, batchNum))
//    env.execute(getClass.getSimpleName)
//  }
//
//  /**
//   *
//   * @param sqlQuery   输入原始查询sql
//   * @param bizDate    替换格式化日期参数
//   * @param simpleFlag 是否限制limit 1,true：格式化limit 1，flase:不格式limit 1
//   * @return
//   */
//  private def formatSQl(sqlQuery: String, bizDate: String, simpleFlag: Boolean): String = {
//    var sql = sqlQuery
//    if (sqlQuery.contains("${bdp.system.bizdate}")) {
//      sql = sql.replaceAll("\\$\\{bdp.system.bizdate}", bizDate)
//    }
//    if (!simpleFlag) {
//      sql
//    } else if (sqlQuery.contains("limit")) {
//      sql.split("limit ")(0) + "limit 1;"
//    } else if (sqlQuery.contains("LIMIT")) {
//      sql.split("LIMIT ")(0) + "limit 1;"
//    } else {
//      sql + " limit 1;"
//    }
//  }
//
//  private def jdbcMappingBasicType(jdbcType: String): (BasicTypeInfo[_], Int) = {
//    jdbcType match {
//      case "TINYINT" => (BasicTypeInfo.SHORT_TYPE_INFO, Types.TINYINT)
//      case "SMALLINT" => (BasicTypeInfo.SHORT_TYPE_INFO, Types.SMALLINT)
//      case "INT" => (BasicTypeInfo.INT_TYPE_INFO, Types.INTEGER)
//      case "BIGINT" => (BasicTypeInfo.LONG_TYPE_INFO, Types.BIGINT)
//      case "LONG" => (BasicTypeInfo.LONG_TYPE_INFO, Types.BIGINT)
//      case "FLOAT" => (BasicTypeInfo.FLOAT_TYPE_INFO, Types.FLOAT)
//      case "DOUBLE" => (BasicTypeInfo.DOUBLE_TYPE_INFO, Types.DOUBLE)
//      case "DECIMAL" => (BasicTypeInfo.DOUBLE_TYPE_INFO, Types.DOUBLE)
//      case "CHAR" => (BasicTypeInfo.STRING_TYPE_INFO, Types.CHAR)
//      case "VARCHAR" => (BasicTypeInfo.STRING_TYPE_INFO, Types.VARCHAR)
//      case "STRING" => (BasicTypeInfo.STRING_TYPE_INFO, Types.NCHAR)
//      case "DATE" => (BasicTypeInfo.DATE_TYPE_INFO, Types.DATE)
//      case "DATETIME" => (BasicTypeInfo.DATE_TYPE_INFO, Types.DATE)
//      case "BOOLEAN" => (BasicTypeInfo.BOOLEAN_TYPE_INFO, Types.BOOLEAN)
//    }
//  }
//
//  private def odpsDataToNeo4jCQL(nodeName: String, metaData: ResultSetMetaData): String = {
//    val sj = new StringJoiner(",", s"create(n:$nodeName{", "})")
//    for (i <- 1 to metaData.getColumnCount) {
//      sj.add(metaData.getColumnName(i) + ":?")
//    }
//    sj.toString
//  }
//}
//
//class Neo4jOutputFormat(neo4jUrl: String, username: String, password: String, query: String, batchNum: Integer) extends RichOutputFormat[Row] {
//  val logger = LoggerFactory.getLogger(this.getClass)
//
//  private var dbConn: Connection = _
//  private var ps: PreparedStatement = _
//  private var typesArray: Array[Int] = _
//  private var batchCount = 0
//  private var batchInterval = JobConstant.DEFAULT_NEO4J_BATCH_SIZE
//  if (batchNum > 0) {
//    batchInterval = batchNum
//  }
//
//  override def configure(parameters: Configuration): Unit = {}
//
//  override def open(taskNumber: Int, numTasks: Int): Unit = {
//    try {
//      Class.forName(JobConstant.NEO4J_JDBC_DRIVER)
//      dbConn = DriverManager.getConnection(neo4jUrl, username, password)
//      ps = dbConn.prepareStatement(query)
//    } catch {
//      case e: SQLException => throw new IllegalArgumentException("open() failed.", e)
//      case e: ClassNotFoundException => throw new IllegalArgumentException("JDBC driver class not found.", e)
//    }
//  }
//
//  override def writeRecord(row: Row): Unit = {
//    //    if (Objects.isNull(typesArray) && (typesArray.size > 0) && (typesArray.size ne row.getArity)) {
//    //      logger.warn("Column SQL types array doesn't match arity of passed Row! Check the passed array...")
//    //    }
//    try {
//      if (Objects.isNull(typesArray)) {
//        // no types provided
//        for (index <- 0 until row.getArity()) {
//          logger.warn("Unknown column type for column {}. Best effort approach to set its value: {}.", index + 1, row.getField(index));
//          if (row.getField(index).isInstanceOf[java.util.Date]) {
//            ps.setObject(index + 1, new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(row.getField(index)))
//          } else {
//            ps.setObject(index + 1, row.getField(index));
//          }
//        }
//      } else {
//        // types provided
//        for (index <- 0 until row.getArity()) {
//          if (Objects.isNull(row.getField(index))) {
//            ps.setNull(index + 1, typesArray(index))
//          } else {
//            typesArray(index) match {
//              case Types.NULL => ps.setNull(index + 1, typesArray(index))
//              case Types.BOOLEAN => ps.setBoolean(index + 1, row.getField(index).asInstanceOf[Boolean])
//              case Types.BIT => ps.setBoolean(index + 1, row.getField(index).asInstanceOf[Boolean])
//              case Types.CHAR => ps.setString(index + 1, row.getField(index).toString)
//              case Types.NCHAR => ps.setString(index + 1, row.getField(index).toString)
//              case Types.VARCHAR => ps.setString(index + 1, row.getField(index).toString)
//              case Types.LONGVARCHAR => ps.setString(index + 1, row.getField(index).toString)
//              case Types.LONGNVARCHAR => ps.setString(index + 1, row.getField(index).toString)
//              case Types.TINYINT => ps.setByte(index + 1, row.getField(index).asInstanceOf[Byte])
//              case Types.SMALLINT => ps.setShort(index + 1, row.getField(index).asInstanceOf[Short])
//              case Types.INTEGER => ps.setInt(index + 1, row.getField(index).asInstanceOf[Int])
//              case Types.BIGINT => ps.setLong(index + 1, row.getField(index).asInstanceOf[Long])
//              case Types.REAL => ps.setFloat(index + 1, row.getField(index).asInstanceOf[Float])
//              case Types.FLOAT => ps.setFloat(index + 1, row.getField(index).asInstanceOf[Float])
//              case Types.DOUBLE => ps.setDouble(index + 1, row.getField(index).asInstanceOf[Double])
//              case Types.DECIMAL => ps.setBigDecimal(index + 1, row.getField(index).asInstanceOf[java.math.BigDecimal])
//              case Types.NUMERIC => ps.setBigDecimal(index + 1, row.getField(index).asInstanceOf[java.math.BigDecimal])
//              case Types.DATE => {
//                if (row.getField(index).isInstanceOf[java.util.Date]) {
//                  ps.setString(index + 1, new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(row.getField(index)))
//                }
//              }
//              case Types.TIME => ps.setTime(index + 1, row.getField(index).asInstanceOf[Time])
//              case Types.TIMESTAMP => ps.setTimestamp(index + 1, row.getField(index).asInstanceOf[Timestamp])
//            }
//          }
//        }
//      }
//      ps.addBatch()
//      batchCount += 1
//    } catch {
//      case e: Exception => throw new RuntimeException("Preparation of JDBC statement failed.", e);
//    }
//    // execute batch
//    if (batchCount >= batchInterval) {
//      logger.info(s"sync data num:$batchInterval")
//      flush()
//    }
//  }
//
//  def flush(): Unit = {
//    try {
//      ps.executeBatch
//      batchCount = 0
//    } catch {
//      case e: SQLException => throw new RuntimeException("Execution of JDBC statement failed.", e)
//    } finally {
//      ps = dbConn.prepareStatement(query)
//    }
//
//  }
//
//  def getTypesArray(): Array[Int] = {
//    typesArray
//  }
//
//  override def close(): Unit = {
//    if (ps != null) {
//      flush()
//      // close the connection
//      try {
//        ps.close();
//      } catch {
//        case e: Exception => logger.error("JDBC statement could not be closed: " + e.getMessage());
//      } finally {
//        ps = null;
//      }
//    }
//    if (dbConn != null) {
//      try {
//        dbConn.close();
//      } catch {
//        case e: Exception => logger.error("JDBC connection could not be closed: " + e.getMessage());
//      } finally {
//        dbConn = null;
//      }
//    }
//  }
//}
//
